package client

import (
	"encoding/json"
	"github.com/IBM/sarama"
	"listenboos-common/kafka/model"
)

// MsgChatTransferClient  kafka客户端接口
type MsgChatTransferClient interface {
	Push(msg *model.AlbumInfo) error
}
type KafkaProducer struct {
	producer sarama.SyncProducer
}

//type SaramaSyncProducer struct {
//	producer sarama.SyncProducer
//}

//func NewSaramaSyncProducer(producer sarama.SyncProducer) MsgChatTransferClient {
//	return &SaramaSyncProducer{
//		producer: producer,
//	}
//}

func (k *KafkaProducer) Push(msg *model.AlbumInfo) error {

	body, err := json.Marshal(msg)
	if err != nil {
		return err
	}
	_, _, err = k.producer.SendMessage(&sarama.ProducerMessage{
		Topic: "albumInfo",
		Value: sarama.ByteEncoder(string(body)),
	})
	return err
}

//func (k *SaramaSyncProducer) Push(msg *model.AlbumInfo) error {
//
//	body, err := json.Marshal(msg)
//	if err != nil {
//		return err
//	}
//	_, _, err = k.producer.SendMessage(&sarama.ProducerMessage{
//		Topic: "albumInfo",
//		Value: sarama.ByteEncoder(string(body)),
//	})
//	return err
//}

func NewKafkaProducer(pc sarama.SyncProducer) *KafkaProducer {
	return &KafkaProducer{
		producer: pc,
	}
}
func NewSyncProducer(client sarama.Client) sarama.SyncProducer {
	res, err := sarama.NewSyncProducerFromClient(client)
	if err != nil {
		panic(err)
	}
	return res
}
